package mqtt;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.time.Duration;
import java.util.*;
import java.util.Map.Entry;

/**
 * elapsed,success,failureMessage,bytes,sentBytes,allThreads,host
 * 计算：
 * QPS、接受消息总数、三个档位的AvgDelay、并发用户数、错误类型统计&计数、发送接受速率
 * <p>
 * 使用Stream实现解析器
 * 采样间隔通过流操作自身的窗口功能实现
 * 计算平均响应时延是采用commons中的DescriptiveStatistics实现
 *
 * @author pf
 */
public class MqttSamplerAnalyse implements Runnable {

	private String brokerUrl;
	private String topic;
	private int interval;
	public MqttSamplerAnalyse(String brokerUrl, String topic, int interval) {

		this.brokerUrl = brokerUrl;
		this.topic = topic;
		this.interval = interval;
	}

	public static void main(String[] args) {

		if (args.length == 3) {
			new Thread(new MqttSamplerAnalyse(args[0], args[1], Integer.valueOf(args[2]))).start();
		} else {
			new Thread(new MqttSamplerAnalyse("172.18.84.143:9092", "test", 10)).start();
		}
	}

	/**
	 * @param brokerUrl kafka连接地址
	 * @param topic     topic名称
	 * @param interval  采样间隔，单位：秒
	 */
	@Override
	public void run() {

		Properties props = new Properties();

		props.put(StreamsConfig.APPLICATION_ID_CONFIG, UUID.randomUUID().toString());
		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

		StreamsBuilder builder = new StreamsBuilder();

		KStream<String, String> textLines = builder.stream(topic);

		textLines.process(() -> new MqttAnalyseProcess(interval));

		KafkaStreams streams = new KafkaStreams(builder.build(), props);
		streams.start();
	}
}

class MqttAnalyseProcess implements Processor<String, String> {

	private final static double LEVEL_1_PERCENT = 70;
	private final static double LEVEL_2_PERCENT = 90;
	private final static double LEVEL_3_PERCENT = 99;
	// 总计数
	private long totalCount = 0L;
	// 记录错误原因
	private TreeMap<String, Integer> failMessage = new TreeMap<String, Integer>();
	private ProcessorContext context;
	private int interval;
	private HashMap<String, ArrayList<JSONObject>> samplers = new HashMap<String, ArrayList<JSONObject>>();
	private BufferedWriter writer;

	public MqttAnalyseProcess(int interval) {

		this.interval = interval;

		try {
			writer = new BufferedWriter(new FileWriter(new File("log.csv")));
			writer.write("timeStamp,qps,avgDelay,level_1_percent,level_2_percent,level_3_percent,maxDelay,minDelay,successCount,failCount,threads,send,receive,totalCount\n");
			writer.flush();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	@Override
	public void init(ProcessorContext context) {

		this.context = context;

		this.context.schedule(Duration.ofSeconds(interval), PunctuationType.STREAM_TIME, (timeStamp) -> {
			analyse(timeStamp);
		});
	}


	@Override
	public void process(String key, String value) {

		JSONObject valueJson = JSONObject.parseObject(value);

		ArrayList<JSONObject> tempList = samplers.get(key);
		if (tempList == null) {
			tempList = new ArrayList<JSONObject>();
		}

		tempList.add(valueJson);
		samplers.put(key, tempList);
	}

	@Override
	public void close() {
	}

	public void analyse(long timeStamp) {

		for (Entry<String, ArrayList<JSONObject>> entry : samplers.entrySet()) {

			ArrayList<JSONObject> value = entry.getValue();

			int tempSuccessCount = 0;
			int tempFailCount = 0;
			// 用来计算Delay的科学计算
			DescriptiveStatistics tempDelayStatistics = new DescriptiveStatistics();
			long tempSendBytes = 0L;
			long tempReceiveBytes = 0L;

			int threads = 0;
			TreeMap<String, Integer> threadsMap = new TreeMap<String, Integer>();

			for (JSONObject jsonObject : value) {

				totalCount++;
				if (jsonObject.getBooleanValue("success")) {
					tempSuccessCount++;
				} else {
					tempFailCount++;
					Integer failCount = failMessage.get(jsonObject.getString("failureMessage"));
					if (failCount == null) {
						failCount = new Integer(0);
						failMessage.put(jsonObject.getString("failureMessage"), failCount);
					}
					failCount++;
				}
				tempDelayStatistics.addValue(jsonObject.getDoubleValue("elapsed"));
				tempSendBytes = tempSendBytes + jsonObject.getLongValue("sentBytes");
				tempReceiveBytes = tempReceiveBytes + jsonObject.getLongValue("bytes");

				threadsMap.put(jsonObject.getString("host"), jsonObject.getInteger("allThreads"));
			}

			for (int tempThreads : threadsMap.values()) {
				threads = threads + tempThreads;
			}

			StringBuffer stringBuffer = new StringBuffer();

			try {
				stringBuffer.append(timeStamp);
				stringBuffer.append(",");
				stringBuffer.append((tempSuccessCount + tempFailCount) / interval);
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getSum() / (tempSuccessCount + tempFailCount));
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getPercentile(LEVEL_1_PERCENT));
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getPercentile(LEVEL_2_PERCENT));
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getPercentile(LEVEL_3_PERCENT));
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getMax());
				stringBuffer.append(",");
				stringBuffer.append(tempDelayStatistics.getMin());
				stringBuffer.append(",");
				stringBuffer.append(tempSuccessCount);
				stringBuffer.append(",");
				stringBuffer.append(tempFailCount);
				stringBuffer.append(",");
				stringBuffer.append(threads);
				stringBuffer.append(",");
				stringBuffer.append(tempSendBytes / interval);
				stringBuffer.append(",");
				stringBuffer.append(tempReceiveBytes / interval);
				stringBuffer.append(",");
				stringBuffer.append(totalCount);
				stringBuffer.append("\n");

				System.out.println(stringBuffer.toString());

				writer.write(stringBuffer.toString());
				writer.flush();
			} catch (Exception e) {
				e.printStackTrace();
			}

		}

		samplers.clear();

	}

}
